package syM

import (
	"db2s/Er"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/outPut"
	"errors"
	"fmt"
	"github.com/google/uuid"
	//"db2s/Fp"
	mq "db2s/topic-mq"
	"os"
	"reflect"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// 每个表的mq消息状态监听，结束时发送状态给bar
func mqTableStatusMonitor(done chan struct{}, wg *sync.WaitGroup, plan *SchedulePlan2, logSeq int64) {
	var (
		event = "[mqTableStatusMonitor]"
		vlog  string
	)
	defer func() {
		wg.Done()
		//terminalPods <- plan.TPod
		newTer.Write(plan.TPod)
		for _, v := range []*os.File{
			//plan.Object.DataFixFile.CreateFilePoint,
			//plan.Object.DataFixFile.AlterFilePoint,
		} {
			if err := v.Close(); err != nil {
				return
			}
		}
		//for _, v := range []string{plan.Object.DataFixFile.CreateFileName, plan.Object.DataFixFile.AlterFileName} {
		//	if _, err := global.FileIsExtRemove(v); err != nil {
		//		log.MainLog().Warn(fmt.Sprintf("(%v) %v delete file %v fail. error info is {%v}", logSeq, event, v, err))
		//		return
		//	}
		//}
	}()
	vlog = fmt.Sprintf("(%d) %v Start monitoring the mq queue status of table %v.%v. The default timeout is 360 days.", logSeq, event, plan.Object.Schema, plan.Object.Table)
	log.MainLog().Debug(vlog)
	for {
		select {
		case d1, ok := <-plan.mqProductCustomerMonitor.ProductMessage:
			if !ok {
				if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
					vlog = fmt.Sprintf("(%v) %v An error occurs during the execution of table %v.%v, and error handling is performed.", logSeq, event, plan.Object.Schema, plan.Object.Table)
					log.MainLog().Warn(vlog)
					if err := xlsWriterTerminalData(plan, logSeq); err != nil {
						return
					}
					PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "error")
					return
				}
				if *plan.mqProductCustomerMonitor.mqProductSeq == *plan.mqProductCustomerMonitor.mqCustomerSeq {
					vlog = fmt.Sprintf("(%v) %v It is detected that the message queue production quantity and consumption quantity in table %v.%v are equal.", logSeq, event, plan.Object.Schema, plan.Object.Table)
					log.MainLog().Debug(vlog)
					st := getTableName(plan.Object.Schema, plan.Object.Table)
					if err := writeBreakPointData(GlobalPConfigs.FilePoint.BreakPointFin, []*string{&st}, logSeq); err != nil {
						PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "finish")
						return
					}
					if err := xlsWriterTerminalData(plan, logSeq); err != nil {
						PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "finish")
						return
					}
					PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "finish")
					vlog = fmt.Sprintf("(%v) %v Table %v.%v data migration is completed.", logSeq, event, plan.Object.Schema, plan.Object.Table)
					log.MainLog().Info(vlog)
					return
				}
			} else {
				for _, v := range d1 {
					atomic.AddInt64(plan.mqProductCustomerMonitor.mqProductSeq, int64(len(v)))
				}
			}
		case <-done:
			close(plan.mqProductCustomerMonitor.ProductMessage)
			vlog = fmt.Sprintf("(%v) %v It is detected that there is no update in the message queue of table %v.%v within 5 seconds, and the queue writing is closed.", logSeq, event, plan.Object.Schema, plan.Object.Table)
			log.MainLog().Debug(vlog)
		}
	}
}

// 结束标记管道
func expensiveOperation(wg *sync.WaitGroup, message mq.Message, done chan struct{}, logseq int64) {
	var (
		plan  = message.Properties["plan"].(*SchedulePlan2)
		event = "[expensiveOperation]"
		vlog  string
	)
	defer func() {
		wg.Done()
	}()
	vlog = fmt.Sprintf("(%d) %v %v Start monitoring table %v.%v tasks and when to start the exit mechanism.", logseq, event, plan.Object.Schema, plan.Object.Table)
	log.MainLog().Debug(vlog)
	ticker := time.NewTicker(global.MonitorInterval * time.Second)
	for {
		select {
		case <-ticker.C:
			if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
				done <- struct{}{}
				return
			}
			//if atomic.LoadInt64(plan.mqProductCustomerMonitor.mqStartTiming) == 1 && atomic.LoadInt64(plan.mqProductCustomerMonitor.mqLevelProductSend) == atomic.LoadInt64(plan.mqProductCustomerMonitor.mqLevelCustomerReceive) {
			//	done <- struct{}{}
			//	vlog = fmt.Sprintf("(%d) %v %v Table %v.%v starts to start the exit task detection mechanism.", logseq, event, plan.Object.Schema, plan.Object.Table)
			//	log.MainLog().Debug(vlog)
			//	return
			//}
		}
	}
}

func writeBreakPointData(fin *os.File, s any, logseq int64) error {
	WriteTextI := outPut.FileOut("text", map[string]any{"fileN": fin, "logSeq": logseq,
		"bufSize": 1024 * 1024 * 4, "sqlType": "sql", "rowsS": "\n",
	})
	if !WriteTextI.AppendWrite("", s) {
		err := errors.New(fmt.Sprintf("%v", `write data fail!`))
		return err
	}
	return nil
}

// 循环读取管道并发送消息
func chanMonitorSendMsg(s chan any, checkMode CheckMode, topic Topic, tableName string, f sendTopicF, sm map[string]any, fatherSonTags map[string][]string, e sendEndAction) bool {
	var (
		uid uuid.UUID
	)
	for {
		select {
		case c, ok := <-s:
			if !ok {
				e()
				return true
			} else {
				var ms = make(map[string]any)
				uid = uuid.New()
				for k, v := range sm {
					if strings.EqualFold(k, "sendMsgKey") {
						ms[v.(string)] = c
					} else {
						ms[k] = v
					}
				}
				//标签生成：
				if v1, ok1 := ms["label"]; ok1 { //标签标识
					for k2 := range v1.(map[string]any) { //循环集合
						if strings.EqualFold(k2, "no") { //first label
							ms["label"] = map[string]any{uid.String(): []string{}}
						} else if len(k2) > 0 { //子级标签
							if v4, ok4 := fatherSonTags[k2]; ok4 {
								v4 = append(v4, uid.String())
								fatherSonTags[k2] = v4
							} else {
								fatherSonTags[k2] = []string{uid.String()}
							}
							ms["label"] = map[string]any{k2: uid.String()}
						}
					}
				}
				if !f(checkMode, topic, tableName, ms) {
					return false
				}
			}
		}
	}
}

// InitToSchedulePlan2Table 初始化 执行计划 以 map 的形式 key 表名 value SchedulePlan2
func InitToSchedulePlan2Table_bak() (ChanSchedulePlanTable chan any) {
	var f2 = global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
	ChanSchedulePlanTable = make(chan any, f2.TableSum)
	global.MyWait.Add(1)
	go func() {
		defer func() {
			global.MyWait.Done()
			close(ChanSchedulePlanTable)
		}()
		for {
			select {
			case x, y := <-f2.ChanSchemaContext:
				if !y {
					return
				}
				if x == nil {
					continue
				}
				if funcName, ok := ScheduleTaskInitFuncRef[GlobalPConfigs.rules.CheckMode]; ok {
					if funcName.Kind() == reflect.Func {
						arguments := []reflect.Value{
							reflect.ValueOf(x)}
						m1 := funcName.Call(arguments)
						var m2 *SchedulePlan2
						for i := 0; i < len(m1); i++ {
							m2 = m1[i].Interface().(*SchedulePlan2)
						}
						ChanSchedulePlanTable <- m2
					}
				}
			}
		}
	}()
	return
}

func initObjectSchedulePlan_bak(s *SchemaInitContext) (plan2 *SchedulePlan2) {
	plan2 = &SchedulePlan2{
		Object: global.Object{TaskSeq: s.Seq, ObjectLogSeq: s.ObjectLogSeq, Schema: s.Schema,
			Table: s.Table, MSchema: s.MSchema, MTable: s.MTable,
			Role: s.Role,
		},
		XlsResult:   initXlsResult(s),
		XlsMiss:     initXlsMiss(s),
		XlsTerminal: initXlsTerminal(s),
	}
	if s.TPod.TableMiss {
		plan2.TPod = s.TPod
		plan2.Object.TableMiss = true
		return
	}
	tailMessageInit(plan2)
	tableStatusInit(plan2)
	subTaskInit(plan2)
	if strings.EqualFold(GlobalPConfigs.repair.DataFix, "file") {
		InitDataFixFile(plan2)
	}
	plan2.TPod = s.TPod
	return plan2
}

func initStructSchedulePlan_bak(s *SchemaInitContext) (plan2 *SchedulePlan2) {
	plan2 = &SchedulePlan2{
		Object: global.Object{TaskSeq: s.Seq, ObjectLogSeq: s.ObjectLogSeq, Schema: s.Schema,
			Table: s.Table, MSchema: s.MSchema, MTable: s.MTable,
			Role:  s.Role,
			Input: global.StartPartConfigInputP{SqlMode: GlobalPConfigs.rules.Struct.SqlMode},
		},
		XlsResult:   initXlsResult(s),
		XlsMiss:     initXlsMiss(s),
		XlsSwitch:   initXlsSwitch(s),
		XlsTerminal: &outPut.WorkSheetTerminal{},
	}
	if s.TPod.TableMiss {
		plan2.TPod = s.TPod
		plan2.Object.TableMiss = true
		return
	}
	tailMessageInit(plan2)
	tableStatusInit(plan2)
	subTaskInit(plan2)
	if strings.EqualFold(GlobalPConfigs.repair.DataFix, "file") {
		InitDataFixFile(plan2)
	}
	plan2.TPod = s.TPod
	return plan2
}

func tailMessageInit(plan2 *SchedulePlan2) {
	//var (
	//	product  int64 = 0
	//	customer int64 = 0
	//	mqStart  int64 = 0
	//	mqLevels int64 = 0
	//	mqLevel  int64 = 0
	//	message        = make(chan map[string][]string)
	//)
	//plan2.mqProductCustomerMonitor = mqProductCustomerMonitorMessage{
	//	mqProductSeq:           &product,
	//	mqCustomerSeq:          &customer,
	//	mqStartTiming:          &mqStart,
	//	mqLevelProductSend:     &mqLevels,
	//	mqLevelCustomerReceive: &mqLevel,
	//	ProductMessage:         message,
	//}
}

func structIsBaseQueryListeners_bak(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName     = string(message.Body)
		plan          = message.Properties["plan"].(*SchedulePlan2)
		event         = "[GetTableExistListeners]"
		topic         = StructIsColumnMeta
		wg            sync.WaitGroup
		uid           uuid.UUID
		tableComment  = make(map[string]global.TableRecordMeta)
		tableMeta     = &global.GetTableMeta{}
		done          = make(chan struct{})
		dTopic        = StructIsBaseQuery
		deleteMapInfo = map[string]any{"dTopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	wg.Add(2)
	plan.subTaskBeginTime = time.Now().UnixNano()
	go expensiveOperation(&wg, message, done, plan.Object.ObjectLogSeq)
	go mqTableStatusMonitor(done, &wg, plan, plan.Object.ObjectLogSeq)
	if GlobalPConfigs.result.Teletypewriter == "bar" {
		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "running")
		log.MainLog().Info(fmt.Sprintf("(%d) %s Add task Bar Run object %v successfully", plan.Object.ObjectLogSeq, event, tableName))
	}
	//判断目标端表是否存在，存在走alter，不存在走create
	//if b, err := plan.GetTableExist(); err != nil {
	//	plan.errorAction(plan.Object.ObjectLogSeq, err)
	//	PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	//	return mq.ReconsumeLater
	//} else {
	//	tableMeta.AlterStatus = b
	//}
	//获取表备注
	//var err error
	//if tableComment["source"], err = plan.GetTableRecordMeta(GlobalPConfigs.SDB, GlobalPConfigs.dSns.SrcDBName, "source", plan.Object.ObjectLogSeq); err != nil {
	//	plan.errorAction(plan.Object.ObjectLogSeq, err)
	//	PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	//	return mq.ReconsumeLater
	//}
	//if tableMeta.AlterStatus {
	//	if tableComment["dest"], err = plan.GetTableRecordMeta(GlobalPConfigs.DDB, GlobalPConfigs.dSns.DestDBName, "dest", plan.Object.ObjectLogSeq); err != nil {
	//		plan.errorAction(plan.Object.ObjectLogSeq, err)
	//		PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	//		return mq.ReconsumeLater
	//	}
	//}
	tableMeta.Comment = tableComment
	uid = uuid.New()
	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"plan": message.Properties["plan"],
		"err": plan.Status.ErrEnd, "label": map[string]any{uid.String(): []string{}}, "tableMeta": tableMeta,
		"subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
		return mq.ReconsumeLater
	}
	atomic.SwapInt64(plan.mqProductCustomerMonitor.mqStartTiming, 1)
	atomic.SwapInt64(plan.mqProductCustomerMonitor.mqLevelProductSend, 1)
	wg.Wait()
	return mq.ConsumeSuccess
}

// GetTableMetaListeners StructIsColumnMeta  监听执行器
func GetTableMetaListeners_bak(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName   = string(message.Body)
		plan        = message.Properties["plan"].(*SchedulePlan2)
		tableMeta   = message.Properties["tableMeta"].(*global.GetTableMeta)
		checkStatus bool
		//err           error
		topic         = StructIsIndexMeta
		uid           uuid.UUID
		uidMap        = make(map[string]any)
		fatherSonTags = make(map[string][]string)
		dTopic        = StructIsColumnMeta
		deleteMapInfo = map[string]any{"dTopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	for _, v := range GlobalPConfigs.structs.AlterObject {
		if strings.EqualFold(v, "column") {
			checkStatus = true
			break
		}
	}
	for k := range message.Properties["label"].(map[string]any) {
		uid = uuid.New()
		uidMap[k] = uid.String()
		fatherSonTags[k] = []string{uid.String()}
	}
	var cm = make(map[string][]global.TableMeta)
	if checkStatus {
		//if cm["source"], err = plan.GetTableColumn(GlobalPConfigs.SDB, GlobalPConfigs.dSns.SrcDBName, "source", plan.Object.ObjectLogSeq); err != nil {
		//	plan.errorAction(plan.Object.ObjectLogSeq, err)
		//	PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
		//	return mq.ReconsumeLater
		//}
		//if tableMeta.AlterStatus {
		//	if cm["dest"], err = plan.GetTableColumn(GlobalPConfigs.DDB, GlobalPConfigs.dSns.DestDBName, "dest", plan.Object.ObjectLogSeq); err != nil {
		//		plan.errorAction(plan.Object.ObjectLogSeq, err)
		//		PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
		//		return mq.ReconsumeLater
		//	}
		//}
	}
	tableMeta.ColumnMeta = cm
	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"plan": message.Properties["plan"],
		"tableMeta": tableMeta, "err": plan.Status.ErrEnd, "label": uidMap,
		"subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
		return mq.ReconsumeLater
	}
	if atomic.LoadInt64(plan.Status.ErrEnd) > -1 {
		plan.mqProductCustomerMonitor.ProductMessage <- fatherSonTags
		atomic.SwapInt64(plan.mqProductCustomerMonitor.mqLevelCustomerReceive, 1)
	}
	return mq.ConsumeSuccess
}

func GetTableIndexListeners_bak(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName = string(message.Body)
		plan      = message.Properties["plan"].(*SchedulePlan2)
		tableMeta = message.Properties["tableMeta"].(*global.GetTableMeta)
		//err           error
		checkStatus   bool
		topic         = StructIsPartMeta
		dTopic        = StructIsIndexMeta
		deleteMapInfo = map[string]any{"dTopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	for _, v := range GlobalPConfigs.structs.AlterObject {
		if strings.EqualFold(v, "index") {
			checkStatus = true
			break
		}
	}
	var im = make(map[string]global.IndexMerge)
	if checkStatus {
		//if im["source"], err = plan.GetTableIndex(GlobalPConfigs.SDB, GlobalPConfigs.dSns.SrcDBName, "source", plan.Object.ObjectLogSeq); err != nil {
		//	plan.errorAction(plan.Object.ObjectLogSeq, err)
		//	PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
		//	return mq.ReconsumeLater
		//}
		//if tableMeta.AlterStatus {
		//	if im["dest"], err = plan.GetTableIndex(GlobalPConfigs.DDB, GlobalPConfigs.dSns.DestDBName, "dest", plan.Object.ObjectLogSeq); err != nil {
		//		plan.errorAction(plan.Object.ObjectLogSeq, err)
		//		PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
		//		return mq.ReconsumeLater
		//	}
		//}
	}
	tableMeta.IndexMeta = im
	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"plan": message.Properties["plan"],
		"tableMeta": tableMeta, "err": plan.Status.ErrEnd, "label": message.Properties["label"],
		"subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}

func GetTablePartitionListeners_bak(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName   = string(message.Body)
		plan        = message.Properties["plan"].(*SchedulePlan2)
		tableMeta   = message.Properties["tableMeta"].(*global.GetTableMeta)
		checkStatus bool
		//err           error
		topic         = StructIsMetaComparison
		dTopic        = StructIsPartMeta
		deleteMapInfo = map[string]any{"dTopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	for _, v := range GlobalPConfigs.structs.AlterObject {
		if strings.EqualFold(v, "partition") {
			checkStatus = true
			break
		}
	}
	var pm = make(map[string][]global.TablePartitionConfig)
	if checkStatus {
		//if pm["source"], err = plan.GetTablePartition(GlobalPConfigs.SDB, GlobalPConfigs.dSns.SrcDBName, "source", plan.Object.ObjectLogSeq); err != nil {
		//	plan.errorAction(plan.Object.ObjectLogSeq, err)
		//	PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
		//	return mq.ReconsumeLater
		//}
		//if tableMeta.AlterStatus {
		//	if pm["dest"], err = plan.GetTablePartition(GlobalPConfigs.DDB, GlobalPConfigs.dSns.DestDBName, "dest", plan.Object.ObjectLogSeq); err != nil {
		//		plan.errorAction(plan.Object.ObjectLogSeq, err)
		//		PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
		//		return mq.ReconsumeLater
		//	}
		//}

	}
	tableMeta.PartitionMeta = pm
	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"plan": message.Properties["plan"],
		"tableMeta": tableMeta, "err": plan.Status.ErrEnd, "label": message.Properties["label"],
		"subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}

func TableAbnormalDataFixSqlListeners_bak(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		tableName     = string(message.Body)
		plan          = message.Properties["plan"].(*SchedulePlan2)
		tableMeta     = message.Properties["tableMeta"].(*global.GetTableMeta)
		topic         = StructIsApply
		dTopic        = StructIsFix
		result        []FixSqlResult
		err           error
		deleteMapInfo = map[string]any{"dTopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	if tableMeta.AlterStatus { //是校验状态不是create table状态
		plan.TPod.Struct.ExecObject = "alter"
		if result, err = plan.alterFixSql(tableMeta, plan.Object.ObjectLogSeq); err != nil {
			plan.errorAction(plan.Object.ObjectLogSeq, err)
			PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
			return mq.ReconsumeLater
		}
	} else {
		plan.TPod.Struct.ExecObject = "create"
		if result, err = plan.createFixSql_bak(tableMeta, plan.Object.ObjectLogSeq); err != nil {
			plan.errorAction(plan.Object.ObjectLogSeq, err)
			PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
			return mq.ReconsumeLater
		}
	}
	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"plan": message.Properties["plan"],
		"tableMeta": tableMeta, "fixSqlResult": result, "err": plan.Status.ErrEnd, "label": message.Properties["label"],
		"subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func TableAbnormalApplySqlListeners_bak(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan          = message.Properties["plan"].(*SchedulePlan2)
		sqlGarth      = message.Properties["fixSqlResult"].([]FixSqlResult)
		dTopic        = StructIsApply
		deleteMapInfo = map[string]any{"dTopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
	)
	defer deleteTopic(deleteMapInfo)
	defer atomic.AddInt64(plan.mqProductCustomerMonitor.mqCustomerSeq, 1)
	if len(sqlGarth) == 0 {
		return mq.ConsumeSuccess
	}
	if err := plan.execFixSql(message); err != nil {
		plan.errorAction(plan.Object.ObjectLogSeq, err)
		PlanContext.TPodTableStatusSwap(plan.TPod.Status, "abnormal")
		return mq.ReconsumeLater
	}
	vlog := fmt.Sprintf("(%d) The delete repair sql statements of table %s.%s are generated.", plan.Object.ObjectLogSeq, plan.Object.Schema, plan.Object.Table)
	log.MainLog().Debug(vlog)
	return mq.ConsumeSuccess
}

// execFixSql 表结构差异修复语句
func (sp SchedulePlan2) execFixSql_bak(message mq.Message) (err error) {
	var (
		applySqlSlice []string
		event         = "[execFixSql]"
		result        any
		record        string
		t             Er.TableActiveEr
		execRecord    = "success"
		comment       = "执行完成"
		sqlGarth      = message.Properties["fixSqlResult"].([]FixSqlResult)
		tableMeta     = message.Properties["tableMeta"].(*global.GetTableMeta)
	)
	if t, err = Er.TActive(Er.TableActive{
		DBType: func() string {
			if len(GlobalPConfigs.dSns.DestDBName) > 0 {
				return GlobalPConfigs.dSns.DestDBName
			} else {
				return GlobalPConfigs.dSns.SrcDBName
			}
		}(),
	}); err != nil {
		log.MainLog().Error(fmt.Sprintf("%v Get the table wildcard %v under the specified schema %v failed. error is %v", event, sp.Object.Table, sp.Object.Schema, err))
		return err
	}
	//处理应用sql集合
	if strings.EqualFold(GlobalPConfigs.repair.DataFix, "table") {
		for _, v := range sqlGarth {
			sq := v.Sql
			applySqlSlice = append(applySqlSlice, sq)
		}
	} else if strings.EqualFold(GlobalPConfigs.repair.DataFix, "file") {
		headInfo := fmt.Sprintf("/* \n------schema:`%s` table:`%s` ddl ------\n", sp.Object.MSchema, sp.Object.MTable)
		applySqlSlice = append(applySqlSlice, headInfo)
		for _, v := range sqlGarth {
			ab := v.AbInfo
			applySqlSlice = append(applySqlSlice, ab)
		}
		endInfo := fmt.Sprintf("*/\n")
		applySqlSlice = append(applySqlSlice, endInfo)
		for _, v := range sqlGarth {
			sq := v.Sql
			if len(v.Sql) > 0 {
				applySqlSlice = append(applySqlSlice, sq)
			}
		}
	} else if strings.EqualFold(GlobalPConfigs.repair.DataFix, "no") {
		return
	}
	//应用集合sql
	if result, err = t.Alter(global.TableActiveInput{
		LogSeq: sp.Object.ObjectLogSeq,
		//DB:     GlobalPConfigs.DDB,
		TableInfo: global.TableInfo{
			Schema:    sp.Object.MSchema,
			Table:     sp.Object.MTable,
			ShardName: "single",
		},
		Input: global.StartPartConfigInputP{
			SqlLogBin: GlobalPConfigs.structs.SqlLogBin,
		},
		Alter: global.Alter{
			Active:    GlobalPConfigs.repair.DataFix,
			SqlGather: applySqlSlice,
			FilePoint1: func() outPut.FileOperations {
				if tableMeta.AlterStatus {
					return sp.fixFile.AlterFilePoint
				} else {
					return sp.fixFile.CreateFilePoint
				}
			}(),
		},
	}); err != nil {
		PlanContext.TPodTableStatusSwap(sp.TPod.Status, "error")
		record = *result.(*string)
		var rawSql1 []string
		for _, v := range sqlGarth {
			ab := v.AbInfo
			rawSql1 = append(rawSql1, ab)
		}
		var errorAfter, errorBefore bool
		for _, d := range sqlGarth {
			fixSql := d.Sql
			if errorAfter {
				comment = "尚未执行"
				execRecord = "not exec"
			}
			if strings.EqualFold(fixSql, record) {
				errorAfter = true
				errorBefore = true
			}
			if !errorBefore {
				comment = "执行成功"
			}
			if len(d.Sql) == 0 {
				continue
			}
			var behavior = "alter"
			switch {
			case strings.HasPrefix(strings.ToLower(d.Sql), "create"):
				behavior = "create"
				comment = "执行失败"
				execRecord = "fail"
			}
			sp.XlsResult.Data.Struct = getResultSheet(behavior, GlobalPConfigs.repair.DataFix, execRecord, fmt.Sprintf("%v", err), fixSql, strings.Join(rawSql1, "\n"))
			sp.XlsResult.Data.Comment = comment
			newCell.Write(*sp.XlsResult)
		}
		return err
	} else {
		for _, d := range sqlGarth {
			if len(d.Sql) == 0 {
				continue
			}
			var behavior = "alter"
			switch {
			case strings.HasPrefix(strings.ToLower(d.Sql), "create"):
				behavior = "create"
			}
			sp.XlsResult.Data.Struct = getResultSheet(behavior, GlobalPConfigs.repair.DataFix, execRecord, fmt.Sprintf("%v", err), d.Sql, "")
			sp.XlsResult.Data.Comment = comment
			newCell.Write(*sp.XlsResult)
		}
	}
	return
}

func (sp SchedulePlan2) createFixSql_bak(gtm *global.GetTableMeta, logSeq int64) ([]FixSqlResult, error) {
	var (
		result       []FixSqlResult
		createResult global.CreatSqlResult
		fs           any
		event        = "[createFixSql]"
		o            Er.TableColumnEr
		err          error
	)
	if o, err = Er.ColumnData(Er.TableColumn{
		DBType: GlobalPConfigs.dSns.DestDBName,
	}); err != nil {
		log.MainLog().Error(fmt.Sprintf("%v Get the table wildcard %v under the specified schema %v failed. error is %v", event, sp.Object.Table, sp.Object.Schema, err))
		return nil, err
	}
	if fs, err = o.CreateSql(global.GetColumnTypeInput{
		LogSeq: logSeq,
		TableInfo: global.TableInfo{
			Schema: sp.Object.MSchema,
			Table:  sp.Object.MTable,
		},
		Plan: GlobalPConfigs.rules.Struct.AlterSubObject,
		//DB:        GlobalPConfigs.DDB,
		TableMeta: *gtm,
		Input:     sp.Object.Input,
	}); err != nil || fs == nil {
		log.MainLog().Warn(fmt.Sprintf("(%v) %v The table creation statement on the target side of generated table %v.%v  is empty. error info is %v", logSeq, event, sp.Object.Schema, sp.Object.Table, err))
		return result, nil
	} else {
		createResult = fs.(global.CreatSqlResult)
	}
	if len(createResult.Record) > 0 {
		for _, v := range createResult.Record {
			sp.XlsSwitch.Data.Struct = getSwitchSheet("create", "table", v.Object, v.Logo, v.OldValue, v.NewValue, "已自动修复")
			sp.XlsSwitch.Data.Comment = v.Reason
			newCell.Write(*sp.XlsSwitch)
		}
	}
	createAbInfo := sp.createAbInfo(gtm)
	result = append(result, FixSqlResult{AbInfo: createAbInfo, Sql: fmt.Sprintf("%v", createResult.Sql)})
	return result, nil
}
